package mr.diff;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.StringTokenizer;

import mr.CMapper;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import util.DiffComputation;
import util.DiffMessage;
import util.DiffMessage2;
import util.Pair;
import util.Parsable;
import util.ScanDList;
import util.Utilties;

public class DiffStep2<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>>
		extends CMapper<T0, T1, T2, DiffMessage2<T1, T2>> {

	private static final Log LOG = LogFactory.getLog(DiffStep2.class);
	T2 accSum;

	int countMsg, flg;
	T0 insT0;
	private OutputCollector<LongWritable, Text> output;
	T2 pacc;
	ScanDList<T1> ptimes;

	public DiffStep2(DiffComputation<T0, T1, T2> di) {
		super(di);

		otimsAcc = otimes.id();
		accSum = oplus.id();
		pacc = null;
		insT0 = p.getInstanceOfT0();
		countMsg = 0;
		flg = 0;

	}

	@Override
	public void close() throws IOException {

		// add all messages and send them in
		if ((countMsg > 0) && (output != null)) {
			DiffMessage2<T1, T2> sendmsg = new DiffMessage2<T1, T2>(chunkey,
					otimsAcc, accSum);
			addMesg(sendmsg);

			// go to same reducer
			output.collect(new LongWritable(0), new Text(sendmsg.toString()));
			// System.out.println("send msg: " + sendmsg);
			LOG.info("===send msg: " + sendmsg);

		} else if (chunkey >= 0)
			LOG.error("===no msg in DiffStep2!!! ");

	}

	@Override
	public void configure(JobConf job) {

		try {
			ptimes = getCachedMessages(job);
			LOG.info("received msgs num: " + ptimes.size());
		} catch (NoSuchMethodException | SecurityException
				| ClassNotFoundException | InstantiationException
				| IllegalAccessException | IllegalArgumentException
				| InvocationTargetException | IOException e) {
			e.printStackTrace();
		}

	}

	private ScanDList<T1> getCachedMessages(JobConf job)
			throws NoSuchMethodException, SecurityException,
			ClassNotFoundException, InstantiationException,
			IllegalAccessException, IllegalArgumentException,
			InvocationTargetException, IOException {

		ScanDList<T1> rst = new ScanDList<T1>();
		Path[] paths = DistributedCache.getLocalCacheFiles(job);
		DiffMessage<T1> receiver = new DiffMessage<T1>(0, otimsAcc);
		for (Path pt : paths) {
			BufferedReader reader = new BufferedReader(new FileReader(
					pt.toString()));
			try {

				String line = null;
				while ((line = reader.readLine()) != null) {
					String partialSumList = line.split("\t")[1];
					partialSumList.trim();
					// System.out.format("read partialSumList: %s%n",
					// partialSumList);
					if (partialSumList.length() > 0) {
						String[] tmp = partialSumList.split(ELE_Delim);
						for (String val : tmp) {
							if (val != null && val.length() > 0) {
								Pair<Long, T1> parsed = receiver.parse(val);
								long key = parsed.getFirst();
								T1 parSum = parsed.getSecond();
								// System.out.format("read key: %d  val: %s%n",
								// key,parSum);
								rst.put(key, parSum); // scan
								countMsg += 1;
							}
						}
					}
				}

			} catch (IOException x) {
				LOG.error("IOException: %s%n", x);
			}
		}

		return rst;
	}

	@Override
	public void map(LongWritable key, Text line,
			OutputCollector<LongWritable, Text> out, Reporter reporter)
			throws IOException {

		// get current chunkey
		if (chunkey < 0) { // only do at very beginning
			chunkey = this.getChunkey(reporter);
			flg++;
			if (ptimes != null) {
				// System.out.println("configure : otimsAcc: " + otimsAcc);
				T1 preAllAcc = Utilties.AccumPrevousTimes(chunkey, ptimes,
						otimes);
				// System.out.println("configure : preAllAcc: " + preAllAcc);
				this.otimsAcc = otimes.evaluate(preAllAcc, this.otimsAcc);  
				if (chunkey == 0)
					this.otimsAcc = otimes.evaluate(cons_c, otimsAcc);
			}
			this.output = out;
		}

		String eles = line.toString().trim();
		if (eles != null && eles.length() > 0) {
			StringTokenizer itr = new StringTokenizer(line.toString().trim(),
					ELE_Delim);
			while (itr.hasMoreTokens()) {
				String tmp = itr.nextToken();
				T0 thisOne = insT0.parse(tmp);
				pacc = p.evaluate(thisOne, otimsAcc);
				accSum = oplus.evaluate(accSum, pacc);
				otimsAcc = otimes.evaluate(otimsAcc, q.evaluate(thisOne));

			}
		}

		reporter.progress();

	}

}